package com.jen.listener;

import cn.hutool.bloomfilter.BitMapBloomFilter;
import cn.hutool.core.date.DateUtil;
import com.jen.queue.DelayedMessageQueueConfig;
import com.jen.service.RabbitSendService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;

/**
 * @ClassName DelayedRabbitListener
 * @Description RabbitMQ监听新生队列消息
 * @Date 2021/12/17 11:46
 * @Author Rick Jen
 */
@Component
@RabbitListener(queues = DelayedMessageQueueConfig.NEWLY_BORN_QUEUE)
@Slf4j
public class DelayedRabbitListener {

    /**
     * 消息的前缀
     */
    private final String MESSAGE = "delayed-message:";

    @RabbitHandler
    @Transactional(rollbackFor = Exception.class)
    public void process(String msg, Message message, Channel channel) throws IOException {
        log.info("接收时间：{},【DelayedRabbitListener-接收到消息】=> {}", DateUtil.now(), msg);
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String messageId = message.getMessageProperties().getMessageId();
        log.info("消息投递ID → ：{}", deliveryTag);
        log.info("消息自定义ID → ：{}", messageId);
        if (this.bitMapBloomFilter.contains(messageId)) {
            log.error("message:{},该消息被消费过，不能重复进行消费", msg);
            try {
                // 如果进入到这里面，说明这个消息之前被消费过，但是 MQ 认为你没有消费，所以我们要签收这条消息
                channel.basicAck(deliveryTag, false);
                return;
            } catch (Exception e) {
                log.error("签收消息异常！异常原因：{}", e.toString());
            }
        }
        // 消息被消费次数统计
        String count = this.redisTemplate.opsForValue().get(MESSAGE + messageId);
        if (count != null && Long.parseLong(count) >= 3) {
            channel.basicNack(deliveryTag, false, false);
            // 失败次数过多，这里可能需要人工介入
            log.error("该消息：{} ->消费【3】次都失败了！", message);
        }
        // 签收消息  最后再签收消息
        channel.basicAck(deliveryTag, true);
        // 消费成功之后放到布隆过滤器里面
        this.bitMapBloomFilter.add(messageId);
        log.info("消息已被签收，消息内容：{}", msg);
    }


    @Autowired
    private BitMapBloomFilter bitMapBloomFilter;

    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * rabbit发送服务
     */
    @Autowired
    private RabbitSendService rabbitSendService;
}
